package com.wlt.service;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.wlt.mapper.SeckillGoodsMapper;
import com.wlt.pojo.CartGoods;
import com.wlt.pojo.Orders;
import com.wlt.pojo.SeckillGoods;
import com.wlt.redis.RedissonLock;
import com.wlt.result.BusException;
import com.wlt.result.CodeEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@DubboService
public class SeckillServiceImpl extends ServiceImpl<SeckillGoodsMapper, SeckillGoods> implements SeckillService
{
    @Autowired
    private SeckillGoodsMapper seckillGoodsMapper;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private RedissonLock redissonLock;
    
    // 黑名单Redis键前缀（确保唯一）
    private static final String BLACKLIST_KEY_PREFIX = "seckill:false_positive:blacklist:";
    // 黑名单过期时间（24小时，可根据业务调整）
    private static final long BLACKLIST_EXPIRE_HOURS = 24;
    // 分布式锁的键前缀
    private static final String LOCK_KEY_PREFIX = "seckill:blacklist:lock:";
    // 定义volatile修饰的布隆过滤器（保证线程可见性）
    private volatile BitMapBloomFilter bitMapBloomFilter = new BitMapBloomFilter(1000);
    
    /**
     * 每分钟查询一次数据库，更新Redis中的秒杀商品数据
     * 条件为startTime < 当前时间 < endTime，库存大于0
     */
    @Scheduled (cron = "0 * * * * *")
    public void refreshRedis () {
        // 1. 将Redis中的库存数据同步到MySQL中
        List<SeckillGoods> seckillGoodsListRedis = redisTemplate.boundHashOps("seckillGoods").values();
        // 避免Redis返回null导致的空指针
        if (seckillGoodsListRedis == null) {
            seckillGoodsListRedis = Collections.emptyList();
        }
        
        // 1.1 批量获取Redis中商品的ID（过滤空数据）
        List<Long> goodsIds = seckillGoodsListRedis.stream()
                                                   .filter(Objects::nonNull) // 过滤null对象
                                                   .map(SeckillGoods::getGoodsId)
                                                   .filter(Objects::nonNull) // 过滤null的goodsId
                                                   .collect(Collectors.toList());
        
        // 1.2 批量查询DB记录（如果没有要更新的ID，直接跳过）
        if (!goodsIds.isEmpty()) {
            List<SeckillGoods> sqlGoodsList = seckillGoodsMapper.selectBatchIds(goodsIds);
            
            // 1.3 批量更新库存（匹配goodsId）
            List<SeckillGoods> finalSeckillGoodsListRedis = seckillGoodsListRedis;
            List<SeckillGoods> updateList = sqlGoodsList.stream()
                                                        .map(sqlGoods -> {
                                                            // 从Redis商品中匹配对应ID的库存
                                                            finalSeckillGoodsListRedis.stream()
                                                                                      .filter(redisGoods -> Objects.equals(redisGoods.getGoodsId(), sqlGoods.getGoodsId()))
                                                                                      .findFirst()
                                                                                      .ifPresent(redisGoods -> sqlGoods.setStockCount(redisGoods.getStockCount()));
                                                            return sqlGoods;
                                                        })
                                                        .collect(Collectors.toList());
            
            // 1.4 批量更新DB（使用Service层的批量更新方法）
            if (!updateList.isEmpty()) {
                // 关键修复：调用IService的批量更新方法（需当前类继承ServiceImpl）
                this.updateBatchById(updateList);
            }
        }
        
        log.info("同步Redis库存到MySQL完成");
        
        // 2. 查询数据库中正在秒杀的商品
        QueryWrapper<SeckillGoods> queryWrapper = new QueryWrapper<>();
        // 修复：使用线程安全的日期处理（Java 8+）
        LocalDateTime now = LocalDateTime.now();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String nowStr = now.format(formatter);
        
        queryWrapper.le("startTime", nowStr)    // 当前时间晚于开始时间
                    .ge("endTime", nowStr)      // 当前时间早于结束时间
                    .gt("stockCount", 0);       // 库存大于0
        
        List<SeckillGoods> seckillGoodsList = seckillGoodsMapper.selectList(queryWrapper);
        // 避免查询结果为null
        if (seckillGoodsList == null) {
            seckillGoodsList = Collections.emptyList();
        }
        
        // 3. 原子替换Redis中的秒杀商品缓存（避免脏数据）
        BoundHashOperations<String, Long, SeckillGoods> tempHash = redisTemplate.boundHashOps("seckillGoods:temp");
        // 先清空临时key（防止残留旧数据）
        tempHash.delete(tempHash.keys());
        // 写入新数据
        seckillGoodsList.forEach(goods -> tempHash.put(goods.getGoodsId(), goods));
        // 原子重命名（旧key会被覆盖，保证缓存一致性）
        redisTemplate.delete("seckillGoods"); // 先删除旧key（避免rename失败）
        redisTemplate.rename("seckillGoods:temp", "seckillGoods");
        
        // 4. 重建布隆过滤器（线程安全处理）
        BitMapBloomFilter newBloomFilter = new BitMapBloomFilter(1000);
        seckillGoodsList.forEach(goods -> newBloomFilter.add(goods.getGoodsId().toString()));
        // 用volatile保证可见性（如果其他线程读取该过滤器）
        this.bitMapBloomFilter = newBloomFilter;
        
        log.info("Redis秒杀商品缓存及布隆过滤器重建完成");
    }
    
    
    /**
     * 查询秒杀商品到一页，用的苞米豆的简单分页模型
     * @param page  页码
     * @param size  每页条数
     * @return      秒杀商品列表
     */
    @Override
    @SentinelResource(defaultFallback = "findPageByRedis", blockHandler = "mySqlBlockHandler")
    public Page<SeckillGoods> findPageByRedis (int page, int size) {
        int start = (page - 1) * size;
        int end = start + size;
        
        // 1. 用HSCAN分页获取商品ID（field）
        ScanOptions options = ScanOptions.scanOptions()
                                         .match("*")       // 匹配所有field（商品ID）
                                         .count(size)               // 每次扫描size个（按需调整）
                                         .build();
        Cursor<Map.Entry<Long, SeckillGoods>> cursor = redisTemplate.boundHashOps("seckillGoods").scan(options);
        
        // 2. 遍历游标，收集指定范围的商品
        List<SeckillGoods> goodsList = new ArrayList<>();
        int currentIndex = 0;
        while (cursor.hasNext() && currentIndex < end) {
            Map.Entry<Long, SeckillGoods> entry = cursor.next();
            if (currentIndex >= start) { // 只保留当前页的商品
                goodsList.add(entry.getValue());
            }
            currentIndex++;
        }
        
        // 3. 获取总数量（Redis Hash的HLEN命令）
        Long total = redisTemplate.boundHashOps("seckillGoods").size();
        
        // 4. 构造分页对象
        Page<SeckillGoods> goodsPage = new Page<>(page, size, total);
        goodsPage.setRecords(goodsList);
        return goodsPage;
    }
    
    
    /**
     * 在Redis缓存内查询秒杀商品 —— 首先要用到布隆过滤器
     * @param goodsId   秒杀商品id
     * @return          返回查询到的商品
     */
    @Override
    public SeckillGoods findSeckillGoodsByRedis (Long goodsId) {
        if (goodsId == null) {
            log.info("商品ID为空");
            
            throw new BusException(CodeEnum.NO_GOODS_ERROR);
        }
        String goodsIdStr = goodsId.toString();
        String blacklistKey = BLACKLIST_KEY_PREFIX + goodsIdStr;
        
        // 1. 先查分布式黑名单（Redis）：存在则直接返回null
        Boolean isInBlacklist = redisTemplate.hasKey(blacklistKey);
        if (Boolean.TRUE.equals(isInBlacklist)) {
            log.info("商品ID在分布式黑名单中，直接返回null，goodsId: {}", goodsId);
            return null;
        }
        
        // 2.布隆过滤器判断商品是否真的存在，如果不存在直接返回
        if (!bitMapBloomFilter.contains(goodsIdStr)) {
            log.info("布隆过滤器判断商品不存在");
            
            throw new BusException(CodeEnum.NO_GOODS_ERROR);
        }
        
        // 3.从Redis中查询秒杀商品 —— 进入至此说明成功通过了布隆过滤器
        SeckillGoods seckillGoods = (SeckillGoods) redisTemplate.boundHashOps("seckillGoods").get(goodsId);
        
        // 4.如果查到商品，返回
        if (seckillGoods != null) {
            // 4.1 Redis内存在
            log.info("从Redis中查询商品");
            
            return seckillGoods;
        } else {
            // 4.2 Redis返回空值，去数据库中查询
            log.info("Redis中没有商品，去数据库中查询");
            
            return null;
        }
    }
    
    /**
     * 创建订单，使用Redisson可重入锁+lua脚本
     * @param orders    秒杀订单数据
     * @return          创建的订单
     */
    @Override
    public Orders createOrder (Orders orders) {
        String lockKey = orders.getCartGoods().get(0).getGoodId().toString();
        
        if (redissonLock.lock(lockKey, 10000L)) {
            try
            {
                // 1.生成订单对象
                orders.setId(IdWorker.getIdStr());      // 手动使用雪花算法生成订单ID（自动生成的会保存到数据库）
                orders.setStatus(1);                    // 订单状态未付款
                orders.setCreateTime(new Date());       // 订单创建时间
                orders.setExpire(new Date(new Date().getTime() + 1000*60*5));   // 订单过期时间
                
                // 计算商品的价格
                CartGoods cartGoods = orders.getCartGoods().get(0);
                Integer num = cartGoods.getNum();
                BigDecimal price = cartGoods.getPrice();
                BigDecimal sum = price.multiply(BigDecimal.valueOf(num));       // 防止精度丢失
                orders.setPayment(sum);

                // 2.减少秒杀商品库存
                // 2.1查询秒杀商品
                SeckillGoods seckillGood = findSeckillGoodsByRedis(cartGoods.getGoodId());
                
                // 2.2查询库存，库存不足抛异常
                Integer stockCount = seckillGood.getStockCount();
                if (stockCount <= 0)
                    throw new BusException(CodeEnum.NO_STOCK_ERROR);

                // 2.3减少库存
                seckillGood.setStockCount(seckillGood.getStockCount() - cartGoods.getNum());
                redisTemplate.boundHashOps("seckillGoods").put(seckillGood.getGoodsId(), seckillGood);

                // 3.生成订单，将订单数据保存到Redis
                redisTemplate.setKeySerializer(new StringRedisSerializer());
                /**
                 * 设置订单5分钟过期，过期后只能拿到Redis内的key，拿不到value，
                 * 而过期时间需要回退商品库存，必须拿到value的商品详情才能拿到数据从而进行回退事件
                 * 所以可以设置一个副本，副本过期时间长于订单时间从而来解决拿不到value的问题
                 */
                redisTemplate.opsForValue().set(orders.getId(), orders, 5, TimeUnit.MINUTES);
                redisTemplate.opsForValue().set(orders.getId() + "_copy", orders, 7, TimeUnit.MINUTES);
                
                log.info("库存还有：{}", seckillGood.getStockCount());

                return orders;
            } finally {
                redissonLock.unlock(lockKey);
            }
        } else {
            // 如果没拿到直接就给线程中断了，其实不会返回null
            return null;
        }
    }
    
    /**
     * 从Redis中查订单
     * @param id    订单id
     * @return      查询的结果
     */
    @Override
    public Orders findOrder (String id) {
        return (Orders) redisTemplate.opsForValue().get(id);
    }
    
    @Override
    public Orders pay (String orderId) {
        // 1.查询订单，设置支付相关数据
        Orders order = findOrder(orderId);
        if (order == null) {
            throw new BusException(CodeEnum.ORDER_EXPIRED_ERROR);
        }
        
        order.setStatus(2);
        order.setPaymentTime(new Date());
        order.setPaymentType(2);            // 支付宝支付
        
        // 2.从Redis中删除数据
        redisTemplate.delete(orderId);
        redisTemplate.delete(orderId + "_copy");
        
        // 3.返回订单数据
        return order;
    }
    
    @Override
    public void addRedisSeckillGoods (SeckillGoods seckillGoods) {
        redisTemplate.boundHashOps("seckillGoods").put(seckillGoods.getGoodsId(), seckillGoods);
        bitMapBloomFilter.add(seckillGoods.getGoodsId().toString());
    }
    
    @Override
    @SentinelResource(value = "findSeckillGoodsByMySql", blockHandler = "mySqlBlockHandler")
    public SeckillGoods findSeckillGoodsByMySql (Long goodsId) {
        // 5.如果没有查到商品，则从数据库中查秒杀商品
        QueryWrapper<SeckillGoods> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("goodsId", goodsId);
        SeckillGoods seckillGoodsMySql = seckillGoodsMapper.selectOne(queryWrapper);
        
        log.info("从MySql中查询商品");
        
        // 6.如果数据库中查不到，返回null
        Date now = new Date();
        if (seckillGoodsMySql == null) {
            String goodsIdStr = goodsId.toString();
            
            addToDistributedBlacklist(goodsIdStr);
            
            throw new BusException(CodeEnum.NO_GOODS_ERROR);
        }
        
        // 7.判断商品是否在秒杀时间内，且库存要大于0
        if (now.after(seckillGoodsMySql.getEndTime()) || now.before(seckillGoodsMySql.getStartTime())) {
            throw new BusException(CodeEnum.WRONG_TIME_ERROR);
        }
        else if (seckillGoodsMySql.getStockCount() <= 0)
        {
            throw new BusException(CodeEnum.WRONG_NUMBER_ERROR);
        }
        
        // 8.如果该商品可以在数据库中查到，说明是Redis数据丢失，将商品保存到Redis并返回该商品
        addRedisSeckillGoods(seckillGoodsMySql);
        
        return seckillGoodsMySql;
    }
    
    /**
     * 将误判ID加入分布式黑名单（Redis），并用Redisson锁保证并发安全，因为只有查询秒杀商品才会使用该方法，所以是private
     * @param goodsIdStr    商品id
     */
    private void addToDistributedBlacklist(String goodsIdStr) {
        String blacklistKey = BLACKLIST_KEY_PREFIX + goodsIdStr;
        String lockKey = LOCK_KEY_PREFIX + goodsIdStr;
        
        // 获取Redisson分布式锁（自动续期，避免锁超时）
        RLock lock = redissonClient.getLock(lockKey);
        try {
            // 尝试加锁，最多等待100ms，10秒后自动释放（根据业务调整）
            boolean locked = lock.tryLock(100, 10000, TimeUnit.MILLISECONDS);
            if (locked) {
                // 加锁成功后再次检查（防止并发场景下重复设置）
                if (!redisTemplate.hasKey(blacklistKey)) {
                    // 存入Redis并设置过期时间
                    redisTemplate.opsForValue().set(blacklistKey, "1",
                                                    BLACKLIST_EXPIRE_HOURS, TimeUnit.HOURS);
                    log.info("商品ID成功加入分布式黑名单，goodsId: {}", goodsIdStr);
                }
            }
        } catch (InterruptedException e) {
            log.error("加入黑名单时获取锁异常", e);
            Thread.currentThread().interrupt();
        } finally {
            // 释放锁（仅释放当前线程持有的锁）
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
    
    /**
     * 手动移除黑名单
     * @param goodsId   商品id
     */
    public void removeFromBlacklist(Long goodsId) {
        if (goodsId != null) {
            String blacklistKey = BLACKLIST_KEY_PREFIX + goodsId.toString();
            redisTemplate.delete(blacklistKey);
            log.info("商品ID从黑名单移除，goodsId: {}", goodsId);
        }
    }
    
    
    /**
     * 降级处理
     * @param goodsId
     * @param blockException
     * @return                  返回空
     */
    public SeckillGoods mySqlBlockHandler (Long goodsId, BlockException blockException) {
        log.info("服务降级方法");
        
        return null;
    }
}
